package rdd.operate;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Int;

import java.util.Arrays;
import java.util.List;

//coalesce、repartition用于改变分区
//有些rdd没有shuffle流程，没有改变分区的能力，可以使用这两个方法进行
public class Spark62_Operate_Partition {
    public static void main(String[] args) {
        final SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("spark");
        final JavaSparkContext jsc = new JavaSparkContext(conf);

        final List<Integer> nums = Arrays.asList(1,2,3,4,5,6,7,8,9,10,11,12,13,14);
        final JavaRDD<Integer> rdd = jsc.parallelize(nums , 2);
        final JavaRDD<Integer> filterRDD = rdd.filter(num -> num % 2 == 0);                     //当前分区为2
        filterRDD.saveAsTextFile("output");
        //coalesce有两个参数，第二个参数代表启不启用shuffle
        //默认不传入是false，代表不开启。不开启就不会对数据进行重新分组，数据仍然在原来的组里，没有起到作用
        //所以coalesce主要用于缩容，如果需要扩容，则需要传入第二个参数
        final JavaRDD<Integer> coalesceRDD = filterRDD.coalesce(3);
        coalesceRDD.saveAsTextFile("output1");
        final JavaRDD<Integer> coalesceRDD1 = filterRDD.coalesce(4,true);
        coalesceRDD1.saveAsTextFile("output2");

        //repartition用于直接改变分区数量，扩容的时候用最合适
        filterRDD.repartition(3).saveAsTextFile("output3");
    }
}
